package org.blame.rpc;

import com.rabbitmq.client.*;
import org.blame.constants.Constants;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class Producer {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory connectionFactory=new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL);

        Connection connection= connectionFactory.newConnection();

        //        2.开启信道
        Channel channel=connection.createChannel();

        channel.queueDeclare(Constants.RPC_REQUEST_QUEUE,true,false,false,null);

        String cotId= UUID.randomUUID().toString();
        String msg="rpc通信通信的  消息的发送";
//        生成一个匿名的队列
        String replyQueueName=channel.queueDeclare().getQueue();

//      消息的发送
        AMQP.BasicProperties properties=new AMQP.BasicProperties()
                .builder()
                .correlationId(cotId)
                .replyTo(replyQueueName)
                .build();
        channel.basicPublish("",Constants.RPC_REQUEST_QUEUE,properties,msg.getBytes(StandardCharsets.UTF_8));

        System.out.println("消息发送成功");
//        阻塞队列，用于存储回调的结果
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

        DefaultConsumer consumer=new DefaultConsumer(channel){

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收到的消息"+new String(body));
                if(cotId.equals(properties.getCorrelationId())){
                    response.offer(new String(body,"UTF-8"));

                }
            }
        };
//        回调消息的接收
        channel.basicConsume(replyQueueName,true,consumer);
//          获得回调的结果
        String take = response.take();
        System.out.println("[RPC客户端接收]"+take);
        channel.close();
        connection.close();
    }
}
